package com.xiaohu.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo2DSLWordCount {
  def main(args: Array[String]): Unit = {
    //创建SparkSession对象
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("DSL语法风格编写spark sql")
      .getOrCreate()

    val df1: DataFrame = sparkSession.read
      .format("csv")
      .schema("line STRING")
      .option("sep", "\n")
      .load("spark/data/wcs/words.txt")

    /**
     * 如果要想使用DSL语法编写spark sql的话，需要导入两个隐式转换
     */
    //将sql中的函数，封装成spark程序中的一个个的函数直接调用，以传参的方式调用
    import org.apache.spark.sql.functions._
    //主要作用是，将来可以在调用的函数中，使用$函数，将列名字符串类型转成一个ColumnName类型
    //而ColumnName是继承自Column类的
    import sparkSession.implicits._


    //老版本聚合操作
    //    df1.select(explode(split($"line","\\|")) as "word")
    //      .groupBy($"word")
    //      .count().show()

    //新版本聚合操作
    val resDF: DataFrame = df1.select(explode(split($"line", "\\|")) as "word")
      .groupBy($"word")
      .agg(count($"word") as "counts")

    resDF.repartition(1)
      .write
      .format("csv")
      .option("sep","\t")
      .mode(SaveMode.Overwrite)
      .save("spark/data/sqlout2")

  }
}
